# This file is based on Lib/test/test_android.py on the CPython main branch.

import ctypes.util
import io
import queue
import re
import subprocess
import sys
import unittest
from array import array
from contextlib import ExitStack, contextmanager
from threading import Thread
from time import time
from unittest.mock import patch

from importlib import import_module
stream_mod_name = (
    "java.android.stream" if sys.version_info < (3, 13) else "_android_support")
TextLogStream = import_module(stream_mod_name).TextLogStream

from ..test_utils import API_LEVEL as api_level, FilterWarningsCase


# (name, level, fileno)
STREAM_INFO = [("stdout", "I", 1), ("stderr", "W", 2)]

# Was `from test.support import LOOPBACK_TIMEOUT`, but Chaquopy doesn't include the
# stdlib `test` module.
LOOPBACK_TIMEOUT = 10.0

redirected_native = False


class TestAndroidOutput(FilterWarningsCase):
    maxDiff = None

    def setUp(self):
        self.logcat_process = subprocess.Popen(
            ["logcat", "-v", "tag"], stdout=subprocess.PIPE,
            errors="backslashreplace"
        )
        self.logcat_queue = queue.Queue()

        def logcat_thread():
            for line in self.logcat_process.stdout:
                self.logcat_queue.put(line.rstrip("\n"))
            self.logcat_process.stdout.close()
        self.logcat_thread = Thread(target=logcat_thread)
        self.logcat_thread.start()

        from ctypes import CDLL, c_char_p, c_int
        android_log_write = getattr(CDLL("liblog.so"), "__android_log_write")
        android_log_write.argtypes = (c_int, c_char_p, c_char_p)
        ANDROID_LOG_INFO = 4

        # Separate tests using a marker line with a different tag.
        tag, message = "python.test", f"{self.id()} {time()}"
        android_log_write(
            ANDROID_LOG_INFO, tag.encode("UTF-8"), message.encode("UTF-8"))
        self.assert_log("I", tag, message, skip=True, timeout=5)

    def assert_logs(self, level, tag, expected, **kwargs):
        for line in expected:
            self.assert_log(level, tag, line, **kwargs)

    def assert_log(self, level, tag, expected, *, skip=False, timeout=0.5):
        deadline = time() + timeout
        while True:
            try:
                line = self.logcat_queue.get(timeout=(deadline - time()))
            except queue.Empty:
                self.fail(f"line not found: {expected!r}")
            if match := re.fullmatch(fr"(.)/{tag}: (.*)", line):
                try:
                    self.assertEqual(level, match[1])
                    self.assertEqual(expected, match[2])
                    break
                except AssertionError:
                    if not skip:
                        raise

    def tearDown(self):
        self.logcat_process.terminate()
        self.logcat_process.wait(LOOPBACK_TIMEOUT)
        self.logcat_thread.join(LOOPBACK_TIMEOUT)

    @contextmanager
    def unbuffered(self, stream):
        stream.reconfigure(write_through=True)
        try:
            yield
        finally:
            stream.reconfigure(write_through=False)

    # In --verbose3 mode, sys.stdout and sys.stderr are captured, so we can't
    # test them directly. Detect this mode and use some temporary streams with
    # the same properties.
    def stream_context(self, stream_name, level):
        # https://developer.android.com/ndk/reference/group/logging
        prio = {"I": 4, "W": 5}[level]

        stack = ExitStack()
        stack.enter_context(self.subTest(stream_name))
        stream = getattr(sys, stream_name)
        native_stream = getattr(sys, f"__{stream_name}__")
        if isinstance(stream, io.StringIO):
            stack.enter_context(
                patch(
                    f"sys.{stream_name}",
                    TextLogStream(
                        prio, f"python.{stream_name}", native_stream.fileno(),
                        errors="backslashreplace"
                    ),
                )
            )
        return stack

    def test_str(self):
        for stream_name, level, fileno in STREAM_INFO:
            with self.stream_context(stream_name, level):
                stream = getattr(sys, stream_name)
                original_stream = getattr(sys, f"__{stream_name}__")
                self.assertIsNot(stream, original_stream)

                # We use assertIn rather than assertEqual, because the demo app wraps the
                # TextLogStream with a ConsoleOutputStream.
                tag = f"python.{stream_name}"
                self.assertIn(f"<TextLogStream '{tag}'>", repr(stream))

                self.assertIs(stream.writable(), True)
                self.assertIs(stream.readable(), False)
                self.assertEqual(stream.fileno(), fileno)
                self.assertEqual("UTF-8", stream.encoding)
                self.assertEqual("backslashreplace", stream.errors)
                self.assertIs(stream.line_buffering, True)
                self.assertIs(stream.write_through, False)

                def write(s, lines=None, *, write_len=None):
                    if write_len is None:
                        write_len = len(s)
                    self.assertEqual(write_len, stream.write(s))
                    if lines is None:
                        lines = [s]
                    self.assert_logs(level, tag, lines)

                # Single-line messages,
                with self.unbuffered(stream):
                    write("", [])

                    write("a")
                    write("Hello")
                    write("Hello world")
                    write(" ")
                    write("  ")

                    # Non-ASCII text
                    write("ol\u00e9")  # Spanish
                    write("\u4e2d\u6587")  # Chinese

                    # Non-BMP emoji
                    write("\U0001f600")

                    # Non-encodable surrogates
                    write("\ud800\udc00", [r"\ud800\udc00"])

                    # Code used by surrogateescape (which isn't enabled here)
                    write("\udc80", [r"\udc80"])

                    # Null characters are logged using "modified UTF-8".
                    write("\u0000", [r"\xc0\x80"])
                    write("a\u0000", [r"a\xc0\x80"])
                    write("\u0000b", [r"\xc0\x80b"])
                    write("a\u0000b", [r"a\xc0\x80b"])

                # Multi-line messages. Avoid identical consecutive lines, as
                # they may activate "chatty" filtering and break the tests.
                write("\nx", [""])
                write("\na\n", ["x", "a"])
                write("\n", [""])
                write("b\n", ["b"])
                write("c\n\n", ["c", ""])
                write("d\ne", ["d"])
                write("xx", [])
                write("f\n\ng", ["exxf", ""])
                write("\n", ["g"])

                # Since this is a line-based logging system, line buffering
                # cannot be turned off, i.e. a newline always causes a flush.
                stream.reconfigure(line_buffering=False)
                self.assertIs(stream.line_buffering, True)

                # However, buffering can be turned off completely if you want a
                # flush after every write.
                with self.unbuffered(stream):
                    write("\nx", ["", "x"])
                    write("\na\n", ["", "a"])
                    write("\n", [""])
                    write("b\n", ["b"])
                    write("c\n\n", ["c", ""])
                    write("d\ne", ["d", "e"])
                    write("xx", ["xx"])
                    write("f\n\ng", ["f", "", "g"])
                    write("\n", [""])

                # "\r\n" should be translated into "\n".
                write("hello\r\n", ["hello"])
                write("hello\r\nworld\r\n", ["hello", "world"])
                write("\r\n", [""])

                # Non-standard line separators should be preserved.
                write("before form feed\x0cafter form feed\n",
                      ["before form feed\x0cafter form feed"])
                write("before line separator\u2028after line separator\n",
                      ["before line separator\u2028after line separator"])

                # String subclasses are accepted, but they should be converted
                # to a standard str without calling any of their methods.
                class CustomStr(str):
                    def splitlines(self, *args, **kwargs):
                        raise AssertionError()

                    def __len__(self):
                        raise AssertionError()

                    def __str__(self):
                        raise AssertionError()

                write(CustomStr("custom\n"), ["custom"], write_len=7)

                # Non-string classes are not accepted.
                for obj in [b"", b"hello", None, 42]:
                    with self.subTest(obj=obj):
                        with self.assertRaisesRegex(
                            TypeError,
                            fr"write\(\) argument must be str, not "
                            fr"{type(obj).__name__}"
                        ):
                            stream.write(obj)

                # Manual flushing is supported.
                write("hello", [])
                stream.flush()
                self.assert_log(level, tag, "hello")
                write("hello", [])
                write("world", [])
                stream.flush()
                self.assert_log(level, tag, "helloworld")

                # Long lines are split into blocks of 1000 characters
                # (MAX_CHARS_PER_WRITE in java/android/stream.py), but
                # TextIOWrapper should then join them back together as much as
                # possible without exceeding 4000 UTF-8 bytes
                # (MAX_BYTES_PER_WRITE).
                #
                # ASCII (1 byte per character)
                write(("foobar" * 700) + "\n",  # 4200 bytes in
                      [("foobar" * 666) + "foob",  # 4000 bytes out
                       "ar" + ("foobar" * 33)])  # 200 bytes out

                # "Full-width" digits 0-9 (3 bytes per character)
                s = "\uff10\uff11\uff12\uff13\uff14\uff15\uff16\uff17\uff18\uff19"
                write((s * 150) + "\n",  # 4500 bytes in
                      [s * 100,  # 3000 bytes out
                       s * 50])  # 1500 bytes out

                s = "0123456789"
                write(s * 200, [])  # 2000 bytes in
                write(s * 150, [])  # 1500 bytes in
                write(s * 51, [s * 350])  # 510 bytes in, 3500 bytes out
                write("\n", [s * 51])  # 0 bytes in, 510 bytes out

    def test_bytes(self):
        for stream_name, level, fileno in STREAM_INFO:
            with self.stream_context(stream_name, level):
                stream = getattr(sys, stream_name).buffer
                tag = f"python.{stream_name}"
                self.assertEqual(f"<BinaryLogStream '{tag}'>", repr(stream))
                self.assertIs(stream.writable(), True)
                self.assertIs(stream.readable(), False)
                self.assertEqual(stream.fileno(), fileno)

                def write(b, lines=None, *, write_len=None):
                    if write_len is None:
                        write_len = len(b)
                    self.assertEqual(write_len, stream.write(b))
                    if lines is None:
                        lines = [b.decode()]
                    self.assert_logs(level, tag, lines)

                # Single-line messages,
                write(b"", [])

                write(b"a")
                write(b"Hello")
                write(b"Hello world")
                write(b" ")
                write(b"  ")

                # Non-ASCII text
                write(b"ol\xc3\xa9")  # Spanish
                write(b"\xe4\xb8\xad\xe6\x96\x87")  # Chinese

                # Non-BMP emoji
                write(b"\xf0\x9f\x98\x80")

                # Null bytes are logged using "modified UTF-8".
                write(b"\x00", [r"\xc0\x80"])
                write(b"a\x00", [r"a\xc0\x80"])
                write(b"\x00b", [r"\xc0\x80b"])
                write(b"a\x00b", [r"a\xc0\x80b"])

                # Invalid UTF-8
                write(b"\xff", [r"\xff"])
                write(b"a\xff", [r"a\xff"])
                write(b"\xffb", [r"\xffb"])
                write(b"a\xffb", [r"a\xffb"])

                # Log entries containing newlines are shown differently by
                # `logcat -v tag`, `logcat -v long`, and Android Studio. We
                # currently use `logcat -v tag`, which shows each line as if it
                # was a separate log entry, but strips a single trailing
                # newline.
                #
                # On newer versions of Android, all three of the above tools (or
                # maybe Logcat itself) will also strip any number of leading
                # newlines.
                write(b"\nx", ["", "x"] if api_level < 30 else ["x"])
                write(b"\na\n", ["", "a"] if api_level < 30 else ["a"])
                write(b"\n", [""])
                write(b"b\n", ["b"])
                write(b"c\n\n", ["c", ""])
                write(b"d\ne", ["d", "e"])
                write(b"xx", ["xx"])
                write(b"f\n\ng", ["f", "", "g"])
                write(b"\n", [""])

                # "\r\n" should be translated into "\n".
                write(b"hello\r\n", ["hello"])
                write(b"hello\r\nworld\r\n", ["hello", "world"])
                write(b"\r\n", [""])

                # Other bytes-like objects are accepted.
                write(bytearray(b"bytearray"))

                mv = memoryview(b"memoryview")
                write(mv, ["memoryview"])  # Continuous
                write(mv[::2], ["mmrve"])  # Discontinuous

                write(
                    # Android only supports little-endian architectures, so the
                    # bytes representation is as follows:
                    array("H", [
                        0,      # 00 00
                        1,      # 01 00
                        65534,  # FE FF
                        65535,  # FF FF
                    ]),

                    # After encoding null bytes with modified UTF-8, the only
                    # valid UTF-8 sequence is \x01. All other bytes are handled
                    # by backslashreplace.
                    ["\\xc0\\x80\\xc0\\x80"
                     "\x01\\xc0\\x80"
                     "\\xfe\\xff"
                     "\\xff\\xff"],
                    write_len=8,
                )

                # Non-bytes-like classes are not accepted.
                for obj in ["", "hello", None, 42]:
                    with self.subTest(obj=obj):
                        with self.assertRaisesRegex(
                            TypeError,
                            fr"write\(\) argument must be bytes-like, not "
                            fr"{type(obj).__name__}"
                        ):
                            stream.write(obj)

    def test_native(self):
        c_write = ctypes.CDLL(ctypes.util.find_library("c")).write
        streams = [("native.stdout", "I", 1), ("native.stderr", "W", 2)]

        def write(b, lines=None):
            self.assertIsInstance(b, bytes)
            self.assertEqual(len(b), c_write(fileno, b, len(b)))
            if lines is None:
                lines = [b.decode()]
            self.assert_logs(level, tag, lines)

        # By default, the native streams are not redirected, so everything written to them
        # will be lost. Because there's no way to undo the redirection, we can only test
        # this mode once per process.
        global redirected_native
        if redirected_native:
            print("Can't re-run non-redirected tests")
        else:
            print("Running non-redirected tests")
            for tag, level, fileno in streams:
                with self.subTest(tag=tag):
                    # Add begin and end markers to make sure these writes don't get mixed
                    # up with the redirected writes below.
                    for b in [
                        b"BEGIN non-redirected tests",
                        b"", b"a", b"Hello", b"Hello world", b"Hello world\n",
                        b"END non-redirected tests"
                    ]:
                        with self.subTest(b=b):
                            write(b, [])

        from com.chaquo.python import Python
        Python.getPlatform().redirectStdioToLogcat()
        redirected_native = True

        for tag, level, fileno in streams:
            with self.subTest(tag=tag):
                write(b"", [])
                write(b"a")
                write(b"Hello")
                write(b"Hello world")

                # See above comment: "Log entries containing newlines..."
                write(b"\n", [""])
                write(b"Hello world\n", ["Hello world"])
                write(b"Hello\nworld", ["Hello", "world"])
                write(b"Hello\nworld\n", ["Hello", "world"])


class TestAndroidRateLimit(unittest.TestCase):
    def test_rate_limit(self):
        # https://cs.android.com/android/platform/superproject/+/android-14.0.0_r1:system/logging/liblog/include/log/log_read.h;l=39
        PER_MESSAGE_OVERHEAD = 28

        # https://developer.android.com/ndk/reference/group/logging
        ANDROID_LOG_DEBUG = 3

        # To avoid flooding the test script output, use a different tag rather
        # than stdout or stderr.
        tag = "python.rate_limit"
        stream = TextLogStream(ANDROID_LOG_DEBUG, tag)

        # Make a test message which consumes 1 KB of the logcat buffer.
        message = "Line {:03d} "
        message += "." * (
            1024 - PER_MESSAGE_OVERHEAD - len(tag) - len(message.format(0))
        ) + "\n"

        # To avoid depending on the performance of the test device, we mock the
        # passage of time.
        mock_now = time()

        def mock_time():
            # Avoid division by zero by simulating a small delay.
            mock_sleep(0.0001)
            return mock_now

        def mock_sleep(duration):
            nonlocal mock_now
            mock_now += duration

        # See java/android/stream.py. The default values of these parameters work
        # well across a wide range of devices, but we'll use smaller values to
        # ensure a quick and reliable test that doesn't flood the log too much.
        MAX_KB_PER_SECOND = 100
        BUCKET_KB = 10
        with patch(f"{stream_mod_name}.MAX_BYTES_PER_SECOND", MAX_KB_PER_SECOND * 1024), \
             patch(f"{stream_mod_name}.BUCKET_SIZE", BUCKET_KB * 1024), \
             patch(f"{stream_mod_name}.sleep", mock_sleep), \
             patch(f"{stream_mod_name}.time", mock_time):
            # Make sure the token bucket is full.
            stream.write("Initial message to reset _prev_write_time")
            mock_sleep(BUCKET_KB / MAX_KB_PER_SECOND)
            line_num = 0

            # Write BUCKET_KB messages, and return the rate at which they were
            # accepted in KB per second.
            def write_bucketful():
                nonlocal line_num
                start = mock_time()
                max_line_num = line_num + BUCKET_KB
                while line_num < max_line_num:
                    stream.write(message.format(line_num))
                    line_num += 1
                return BUCKET_KB / (mock_time() - start)

            # The first bucketful should be written with minimal delay. The
            # factor of 2 here is not arbitrary: it verifies that the system can
            # write fast enough to empty the bucket within two bucketfuls, which
            # the next part of the test depends on.
            self.assertGreater(write_bucketful(), MAX_KB_PER_SECOND * 2)

            # Write another bucketful to empty the token bucket completely.
            write_bucketful()

            # The next bucketful should be written at the rate limit.
            self.assertAlmostEqual(
                write_bucketful(), MAX_KB_PER_SECOND,
                delta=MAX_KB_PER_SECOND * 0.1
            )

            # Once the token bucket refills, we should go back to full speed.
            mock_sleep(BUCKET_KB / MAX_KB_PER_SECOND)
            self.assertGreater(write_bucketful(), MAX_KB_PER_SECOND * 2)


class TestAndroidInput(FilterWarningsCase):

    def test_str(self):
        self.assertIs(sys.stdin, sys.__stdin__)
        self.assertTrue(sys.stdin.readable())
        self.assertFalse(sys.stdin.writable())
        self.assertEqual("", sys.stdin.read())
        self.assertEqual("", sys.stdin.read(42))
        self.assertEqual("", sys.stdin.readline())
        self.assertEqual("", sys.stdin.readline(42))

    def test_bytes(self):
        buffer = sys.stdin.buffer
        self.assertTrue(buffer.readable())
        self.assertFalse(buffer.writable())
        self.assertEqual(b"", buffer.read())
        self.assertEqual(b"", buffer.read(42))
        self.assertEqual(b"", buffer.readline())
        self.assertEqual(b"", buffer.readline(42))
